package org.pentaho.redis.output;

import org.eclipse.swt.widgets.Shell;
import org.pentaho.di.core.CheckResult;
import org.pentaho.di.core.CheckResultInterface;
import org.pentaho.di.core.annotations.Step;
import org.pentaho.di.core.database.DatabaseMeta;
import org.pentaho.di.core.exception.KettleException;
import org.pentaho.di.core.exception.KettleValueException;
import org.pentaho.di.core.exception.KettleXMLException;
import org.pentaho.di.core.injection.InjectionSupported;
import org.pentaho.di.core.row.RowMetaInterface;
import org.pentaho.di.core.variables.VariableSpace;
import org.pentaho.di.core.xml.XMLHandler;
import org.pentaho.di.i18n.BaseMessages;
import org.pentaho.di.repository.ObjectId;
import org.pentaho.di.repository.Repository;
import org.pentaho.di.trans.Trans;
import org.pentaho.di.trans.TransMeta;
import org.pentaho.di.trans.step.*;
import org.pentaho.metastore.api.IMetaStore;
import org.w3c.dom.Node;

import java.util.List;


@Step(
        id = "RedisStep",
        name = "RedisStep.Name",
        description = "RedisStep.TooltipDesc",
        image = "org/pentaho/redis/output/resources/redis.svg",
        categoryDescription = "i18n:org.pentaho.di.trans.step:BaseStep.Category.Output",
        i18nPackageName = "org.pentaho.redis.output",
        documentationUrl = "RedisStep.DocumentationURL"
)
@InjectionSupported(localizationPrefix = "RedisStepMeta.Injection.")
public class RedisStepMeta extends BaseStepMeta implements StepMetaInterface {

    /**
     * The PKG member is used when looking up internationalized strings.
     * The properties file with localized keys is expected to reside in
     * {the package of the class specified}/messages/messages_{locale}.properties
     */
    private static final Class<?> PKG = RedisStepMeta.class; // for i18n purposes

    /**
     * Stores the name of the field added to the row-stream.
     */
    private String redisUrl;
    private String redisKey;
    private Boolean redisKeyInField;
    private String redisKeyField;
    private String redisType;
    private String redisValueField;
    private Long millisecondsToExpire;


    /**
     * Constructor should call super() to make sure the base class has a chance to initialize properly.
     */
    public RedisStepMeta() {
        super();
    }

    /**
     * Called by Spoon to get a new instance of the SWT dialog for the step.
     * A standard implementation passing the arguments to the constructor of the step dialog is recommended.
     *
     * @param shell     an SWT Shell
     * @param meta      description of the step
     * @param transMeta description of the the transformation
     * @param name      the name of the step
     * @return new instance of a dialog for this step
     */
    public StepDialogInterface getDialog(Shell shell, StepMetaInterface meta, TransMeta transMeta, String name) {
        return new RedisStepDialog(shell, meta, transMeta, name);
    }

    /**
     * Called by PDI to get a new instance of the step implementation.
     * A standard implementation passing the arguments to the constructor of the step class is recommended.
     *
     * @param stepMeta          description of the step
     * @param stepDataInterface instance of a step data class
     * @param cnr               copy number
     * @param transMeta         description of the transformation
     * @param disp              runtime implementation of the transformation
     * @return the new instance of a step implementation
     */
    public StepInterface getStep(StepMeta stepMeta, StepDataInterface stepDataInterface, int cnr, TransMeta transMeta,
                                 Trans disp) {
        return new RedisStep(stepMeta, stepDataInterface, cnr, transMeta, disp);
    }

    /**
     * Called by PDI to get a new instance of the step data class.
     */
    public StepDataInterface getStepData() {
        return new RedisStepData();
    }

    public void setDefault() {
        setRedisUrl("redis://127.0.0.1:6379/0");
        setRedisKey("");
        setRedisKeyInField(false);
        setRedisKeyField("");
        setRedisType("List");
        setRedisValueField("");
        setMillisecondsToExpire(0L);
    }

    public String getRedisUrl() {
        return redisUrl;
    }

    public void setRedisUrl(String redisUrl) {
        this.redisUrl = redisUrl;
    }

    public String getRedisKey() {
        return redisKey;
    }

    public void setRedisKey(String redisKey) {
        this.redisKey = redisKey;
    }

    public Boolean getRedisKeyInField() {
        return redisKeyInField;
    }

    public void setRedisKeyInField(Boolean redisKeyInField) {
        this.redisKeyInField = redisKeyInField;
    }

    public String getRedisKeyField() {
        return redisKeyField;
    }

    public String getRedisType() {
        return redisType;
    }

    public void setRedisType(String redisType) {
        this.redisType = redisType;
    }

    public void setRedisKeyField(String redisKeyField) {
        this.redisKeyField = redisKeyField;
    }

    public String getRedisValueField() {
        return redisValueField;
    }

    public void setRedisValueField(String redisValueField) {
        this.redisValueField = redisValueField;
    }

    public Long getMillisecondsToExpire() {
        return millisecondsToExpire;
    }

    public void setMillisecondsToExpire(Long millisecondsToExpire) {
        this.millisecondsToExpire = millisecondsToExpire;
    }

    /**
     * This method is used when a step is duplicated in Spoon. It needs to return a deep copy of this
     * step meta object. Be sure to create proper deep copies if the step configuration is stored in
     * modifiable objects.
     * <p>
     * See org.pentaho.di.trans.steps.rowgenerator.RowGeneratorMeta.clone() for an example on creating
     * a deep copy.
     *
     * @return a deep copy of this
     */
    public Object clone() {
        Object retval = super.clone();
        return retval;
    }

    /**
     * This method is called by Spoon when a step needs to serialize its configuration to XML. The expected
     * return value is an XML fragment consisting of one or more XML tags.
     * <p>
     * Please use org.pentaho.di.core.xml.XMLHandler to conveniently generate the XML.
     *
     * @return a string containing the XML serialization of this step
     */
    public String getXML() throws KettleValueException {
        StringBuilder xml = new StringBuilder();
        // only one field to serialize
        xml.append(XMLHandler.addTagValue("redisUrl", redisUrl));
        xml.append(XMLHandler.addTagValue("redisKey", redisKey));
        xml.append(XMLHandler.addTagValue("redisKeyInField", redisKeyInField));
        xml.append(XMLHandler.addTagValue("redisKeyField", redisKeyField));
        xml.append(XMLHandler.addTagValue("redisType", redisType));
        xml.append(XMLHandler.addTagValue("redisValueField", redisValueField));
        xml.append(XMLHandler.addTagValue("millisecondsToExpire", millisecondsToExpire));
        return xml.toString();
    }


    /**
     * This method is called by PDI when a step needs to load its configuration from XML.
     * <p>
     * Please use org.pentaho.di.core.xml.XMLHandler to conveniently read from the
     * XML node passed in.
     *
     * @param stepnode  the XML node containing the configuration
     * @param databases the databases available in the transformation
     * @param metaStore the metaStore to optionally read from
     */
    public void loadXML(Node stepnode, List<DatabaseMeta> databases, IMetaStore metaStore) throws KettleXMLException {
        try {
            setRedisUrl(pNull(XMLHandler.getNodeValue(XMLHandler.getSubNode(stepnode, "redisUrl"))));
            setRedisKey(pNull(XMLHandler.getNodeValue(XMLHandler.getSubNode(stepnode, "redisKey"))));
            setRedisKeyInField(XMLHandler.getNodeValue(XMLHandler.getSubNode(stepnode, "redisKeyInField")).equals("Y"));
            setRedisKeyField(pNull(XMLHandler.getNodeValue(XMLHandler.getSubNode(stepnode, "redisKeyField"))));
            setRedisType(pNull(XMLHandler.getNodeValue(XMLHandler.getSubNode(stepnode, "redisType"))));
            setRedisValueField(pNull(XMLHandler.getNodeValue(XMLHandler.getSubNode(stepnode, "redisValueField"))));
            setMillisecondsToExpire(Long.parseLong(XMLHandler.getNodeValue(XMLHandler.getSubNode(stepnode, "millisecondsToExpire"))));
        } catch (Exception e) {
            throw new KettleXMLException("Pentaho redis output plugin unable to read step info from XML node", e);
        }
    }

    /**
     * This method is called by Spoon when a step needs to serialize its configuration to a repository.
     * The repository implementation provides the necessary methods to save the step attributes.
     *
     * @param rep               the repository to save to
     * @param metaStore         the metaStore to optionally write to
     * @param id_transformation the id to use for the transformation when saving
     * @param id_step           the id to use for the step  when saving
     */
    public void saveRep(Repository rep, IMetaStore metaStore, ObjectId id_transformation, ObjectId id_step)
            throws KettleException {
        try {
            rep.saveStepAttribute(id_transformation, id_step, "redisUrl", redisUrl);
            rep.saveStepAttribute(id_transformation, id_step, "redisKey", redisKey);
            rep.saveStepAttribute(id_transformation, id_step, "redisKeyInField", redisKeyInField);
            rep.saveStepAttribute(id_transformation, id_step, "redisKeyField", redisKeyField);
            rep.saveStepAttribute(id_transformation, id_step, "redisType", redisType);
            rep.saveStepAttribute(id_transformation, id_step, "redisValueField", redisValueField);
            rep.saveStepAttribute(id_transformation, id_step, "millisecondsToExpire", millisecondsToExpire);
        } catch (Exception e) {
            throw new KettleException("Unable to save step into repository: " + id_step, e);
        }
    }

    /**
     * This method is called by PDI when a step needs to read its configuration from a repository.
     * The repository implementation provides the necessary methods to read the step attributes.
     *
     * @param rep       the repository to read from
     * @param metaStore the metaStore to optionally read from
     * @param id_step   the id of the step being read
     * @param databases the databases available in the transformation
     */
    public void readRep(Repository rep, IMetaStore metaStore, ObjectId id_step, List<DatabaseMeta> databases)
            throws KettleException {
        try {
            redisUrl = rep.getStepAttributeString(id_step, "redisUrl");
            redisKey = rep.getStepAttributeString(id_step, "redisKey");
            redisKeyInField = rep.getStepAttributeBoolean(id_step, "redisKeyInField");
            redisKeyField = rep.getStepAttributeString(id_step, "redisKeyField");
            redisType = rep.getStepAttributeString(id_step, "redisType");
            redisValueField = rep.getStepAttributeString(id_step, "redisValueField");
            millisecondsToExpire = rep.getStepAttributeInteger(id_step, "millisecondsToExpire");
        } catch (Exception e) {
            throw new KettleException("Unable to load step from repository", e);
        }
    }


    /**
     * This method is called when the user selects the "Verify Transformation" option in Spoon.
     * A list of remarks is passed in that this method should add to. Each remark is a comment, warning, error, or ok.
     * The method should perform as many checks as necessary to catch design-time errors.
     * <p>
     * Typical checks include:
     * - verify that all mandatory configuration is given
     * - verify that the step receives any input, unless it's a row generating step
     * - verify that the step does not receive any input if it does not take them into account
     * - verify that the step finds fields it relies on in the row-stream
     *
     * @param remarks   the list of remarks to append to
     * @param transMeta the description of the transformation
     * @param stepMeta  the description of the step
     * @param prev      the structure of the incoming row-stream
     * @param input     names of steps sending input to the step
     * @param output    names of steps this step is sending output to
     * @param info      fields coming in from info steps
     * @param metaStore metaStore to optionally read from
     */
    public void check(List<CheckResultInterface> remarks, TransMeta transMeta, StepMeta stepMeta, RowMetaInterface prev,
                      String[] input, String[] output, RowMetaInterface info, VariableSpace space, Repository repository,
                      IMetaStore metaStore) {
        CheckResult cr;

        // See if there are input streams leading to this step!
        if (input != null && input.length > 0) {
            cr = new CheckResult(CheckResult.TYPE_RESULT_OK,
                    BaseMessages.getString(PKG, "Redis.CheckResult.ReceivingRows.OK"), stepMeta);
            remarks.add(cr);
        } else {
            cr = new CheckResult(CheckResult.TYPE_RESULT_ERROR,
                    BaseMessages.getString(PKG, "Redis.CheckResult.ReceivingRows.ERROR"), stepMeta);
            remarks.add(cr);
        }
    }

    private String pNull(String text) {
        if (text == null) {
            return "";
        } else {
            return text;
        }
    }
}
